package cn.hty.mymsg.handler.deduplication.limit;

import cn.hty.mymsg.common.constant.CommonConstant;
import cn.hty.mymsg.common.domain.TaskInfo;
import cn.hty.mymsg.handler.deduplication.DeduplicationParam;
import cn.hty.mymsg.handler.deduplication.service.AbstractDeduplicationService;
import cn.hty.mymsg.utils.RedisUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 内容去重
 */
@Service(value = "SimpleLimitService")
public class SimpleLimitService extends AbstractLimitService{
    private static final String LIMIT_TAG = "SP_";
    @Autowired
    private RedisUtils redisUtils;
    @Override
    public Set<String> limitFilter(AbstractDeduplicationService service, TaskInfo taskInfo, DeduplicationParam param) {
        //设置一个集合接收过滤者
        Set<String> filterReceiver = new HashSet<>();
        // 获取redis记录
        Map<String, String> readyPutRedisReceiver = new HashMap<>(taskInfo.getReceiver().size());
        //获取keys
        List<String> keys = deduplicationAllKey(service, taskInfo).stream().map(key -> LIMIT_TAG + key).collect(Collectors.toList());
        Map<String, String> inRedisValue = redisUtils.mGet(keys);
        //通过key获取键值对
        for (String receiver : taskInfo.getReceiver()) {
            String key = LIMIT_TAG + deduplicationSingleKey(service, taskInfo, receiver);
            //拿到redis里面的值
            String value = inRedisValue.get(key);

            // 符合条件的用户
            if (Objects.nonNull(value) && Integer.parseInt(value) >= param.getCountNum()) {
                filterReceiver.add(receiver);
            } else {
                readyPutRedisReceiver.put(receiver, key);
            }
        }
// 不符合条件的用户：需要更新Redis(无记录添加，有记录则累加次数)
        putInRedis(readyPutRedisReceiver, inRedisValue, param.getDeduplicationTime());
        return filterReceiver;
    }

    private void putInRedis(Map<String, String> readyPutRedisReceiver,
                            Map<String, String> inRedisValue, Long deduplicationTime){
        Set<Map.Entry<String, String>> entries = readyPutRedisReceiver.entrySet();
        Map<String, String> keyValues = new HashMap<>(readyPutRedisReceiver.size());
        //获取key,判断redis是否有，有就计数加1，没有就更新
        for (Map.Entry<String, String> entry : entries) {
            String key = entry.getValue();
            String value = inRedisValue.get(key);
            if (Objects.nonNull(value)){
                keyValues.put(key, String.valueOf(Integer.parseInt(inRedisValue.get(key)) + 1));
            }else {
                keyValues.put(key, String.valueOf(CommonConstant.TRUE));
            }
            //批量导入redis
            redisUtils.pipelineSetEx(keyValues,deduplicationTime);
        }

    }
}
